//package com.itheima.redis_queue;
//
//import java.util.Map;
//import java.util.Set;
//
//import org.redisson.api.RBlockingQueue;
//import org.redisson.api.RedissonClient;
//import org.springframework.beans.BeansException;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.context.ApplicationContext;
//import org.springframework.context.ApplicationContextAware;
//import org.springframework.stereotype.Component;
//
//import com.alibaba.fastjson.JSON;
//
//import lombok.extern.slf4j.Slf4j;
//
///**
// * 初始化队列监听
// * 如果某个类实现了ApplicationContextAware接口，会在类初始化完成后调用setApplicationContext（）方法进行操作
// * 如果某个类实现了InitializingBean接口，会在类初始化完成后，
// * 并在setApplicationContext（）方法执行完毕后，调用afterPropertiesSet（）方法进行操作
// */
//@Component
//@Slf4j
//public class RedisDelayedQueueInit implements ApplicationContextAware {
//
//    @Autowired
//    RedissonClient redissonClient;
//
//    /**
//     * 获取应用上下文并获取相应的接口实现类, 会在类初始化完成后调用该方法
//     * @param applicationContext
//     * @throws BeansException
//     */
//    @Override
//    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
//        Map<String, RedisDelayedQueueListener> map = applicationContext.getBeansOfType(RedisDelayedQueueListener.class);
////        map.entrySet().stream().forEach(taskEventListenerEntry -> {
////            String listenerName = taskEventListenerEntry.getValue().getClass().getName();
////            startThread(listenerName, taskEventListenerEntry.getValue());
////        });
//        Set<Map.Entry<String, RedisDelayedQueueListener>> entrySet = map.entrySet();
//        for (Map.Entry<String, RedisDelayedQueueListener> entry : entrySet) {
//            String listenerName = entry.getValue().getClass().getName();
//            startThread(listenerName, entry.getValue());
//        }
//    }
//
//    /**
//     * 启动线程获取队列
//     * @param queueName                 队列名称
//     * @param redisDelayedQueueListener 任务回调监听
//     * @param <T>                       泛型
//     */
//    private <T> void startThread(String queueName, RedisDelayedQueueListener redisDelayedQueueListener) {
//        RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
//        // 由于此线程需要常驻状态，可以新建线程，不用交给线程池管理
//        Thread thread = new Thread(() -> {
//            log.info("启动监听队列线程：" + queueName);
//            while (true) {
//                try {
//                    T value = blockingFairQueue.take();
//                    log.info("监听队列线程{}, 获取到值:{}", queueName, JSON.toJSONString(value));
//                    new Thread(() -> redisDelayedQueueListener.invoke(value)).start();
//                } catch (Exception e) {
//                    log.info("监听队列线程异常,", e);
//                    try {
//                        Thread.sleep(10000);
//                    } catch (InterruptedException ex) {
//                        log.error("线程出现异常,", ex);
//                    }
//                }
//            }
//        });
//        thread.setName(queueName);
//        thread.start();
//    }
//}
